

namespace Stee.CAP8.MessageBus.Kafka.Factories
{
    using System;
    using Confluent.Kafka;
    using Stee.CAP8.MessageBus.Kafka.Configurations;
    using Stee.CAP8.MessageBus.Kafka.Constants;
    using Stee.CAP8.MessageBus.Kafka.Utility;

    /// <summary>
    /// Factory to create new Kafka admin client
    /// </summary>
    public static class AdminClientFactory
    {
        /// <summary>
        /// Create new admin client
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <param name="adminErrorHandler">Admin client error handler</param>
        /// <returns>New admin client instance</returns>
        public static IAdminClient CreateNewAdminClient(string jsonConfig, Action<IAdminClient, Error> adminErrorHandler)
        {
            var handler = adminErrorHandler ?? throw new ArgumentNullException(nameof(adminErrorHandler));

            var config = InitializeAdminConfig(jsonConfig);
            var client = BuildAdminClient(config, handler);

            return client;
        }


        #region Private Methods

        /// <summary>
        /// Initialize admin configuration
        /// </summary>
        /// <param name="jsonConfig">JSON configuration file path</param>
        /// <returns>Return new admin client configuration</returns>
        private static AdminClientConfig InitializeAdminConfig(string jsonConfig)
        {
            //// Initialize admin
            var adminConfigs = ConfigurationHelper.GetBrokerConfiguration(jsonConfig, ConfigurationConstants.BrokerConfigAdminSection);
            var configuration = new AdminConfiguration(adminConfigs);

            return new AdminClientConfig
            {
                BootstrapServers = configuration.BootstrapServers,
                //ClientId = configuration.ClientID,
            };
        }

        /// <summary>
        /// Build admin client
        /// </summary>
        /// <param name="adminConfigs">Admin client configurations</param>
        /// <param name="adminErrorHandler">Admin client error handler</param>
        /// <returns>Return new admin client instance</returns>
        private static IAdminClient BuildAdminClient(AdminClientConfig adminConfigs, Action<IAdminClient, Error> adminErrorHandler)
        {
            var adminClientBuilder = new AdminClientBuilder(adminConfigs);
            adminClientBuilder.SetErrorHandler(adminErrorHandler);

            return adminClientBuilder.Build();
        }

        #endregion Private Methods
    }
}
